package cn.doitedu.marketing.engine.querier;

import cn.doitedu.marketing.engine.gauges.AvgGauge;
import cn.doitedu.marketing.engine.gauges.CountGauge;
import cn.doitedu.marketing.engine.gauges.QueryGauges;
import cn.doitedu.marketing.engine.utils.ConfigNames;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;

import java.io.IOException;
import java.util.Map;
import java.util.Set;
@Slf4j
public class HbaseQuerier implements RuleConditionQuerier {
    Connection hbaseConn;
    Config config;
    CountGauge countGauge;
    AvgGauge avgRateGauge;

    public HbaseQuerier(Connection hbaseConn, QueryGauges queryGauges) {
        this.hbaseConn = hbaseConn;
        this.config = ConfigFactory.load();
        countGauge = queryGauges.getHbaseQueryCountGauge();
        avgRateGauge = queryGauges.getHbaseAvgrateGauge();
    }

    public boolean queryProfileCondition(String deviceId, Map<String, String> userProfileConditions) throws IOException {
        Table table = hbaseConn.getTable(TableName.valueOf(config.getString(ConfigNames.HBASE_PROFILE_TABLE)));

        // 设置hbase的查询条件：rowKey
        Get get = new Get(deviceId.getBytes());
        // 设置要查询的family和qualifier（标签名）
        Set<String> tags = userProfileConditions.keySet();
        for (String tag : tags) {
            get.addColumn("f".getBytes(), tag.getBytes());
        }

        // 然后请求hbase查询
        long s = System.currentTimeMillis();
        Result result = table.get(get);
        for (String tag : tags) {
            byte[] value = result.getValue("f".getBytes(), tag.getBytes());
            try {
                if (!userProfileConditions.get(tag).equals(new String(value))) {
                    long e = System.currentTimeMillis();
                    avgRateGauge.inc(e-s);
                    countGauge.inc();
                    log.debug("画像条件匹配失败");
                    return false;
                }
            } catch (Exception e) {
                return false;
            }
        }

        long e = System.currentTimeMillis();
        avgRateGauge.inc(e-s);
        countGauge.inc();
        log.debug("画像条件匹配成功");
        return true;
    }


}
